package com.zhao.task.service;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.springframework.util.CollectionUtils;

import com.google.common.collect.Lists;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.zhao.task.config.TaskConfig;
import com.zhao.task.inter.Consume;
import com.zhao.task.inter.Produce;
import com.zhao.task.inter.TaskAction;

import lombok.extern.slf4j.Slf4j;

/**
 * 轮询服务具体业务实现
 *
 * @author zhao
 * @date 2018-11-07
 */
@Slf4j
@SuppressWarnings("all")
public class TaskService<T> {

	/**
	 * 启停标识
	 */
	private Boolean revStopSign;

	/**
	 * 轮询配置
	 */
	private TaskConfig config;

	/**
	 * 轮询实现生产者抽象接口
	 */
	private Produce produce;

	/**
	 * 生产者线程池
	 */
	private ExecutorService producerExecutor;

	/**
	 * 生产者Future
	 */
	private List<Future> producerFutureList;

	/**
	 * 轮询实现消费者抽象接口
	 */
	private Consume consume;

	/**
	 * 消费者线程池
	 */
	private ExecutorService consumerExecutor;

	/**
	 * 消费者Future
	 */
	private List<Future> consumerFutureList;

	/**
	 * 阻塞队列
	 */
	private BlockingDeque<T> queue;

	public TaskService() {
	}

	public TaskService(TaskConfig config, TaskAction action) {
		this.config = config;
		this.produce = action;
		this.consume = action;
		this.queue = new LinkedBlockingDeque(this.config.getQueueSize());
		// 启停标识默认赋值为false
		this.revStopSign = false;
	}

	/**
	 * 启动轮询
	 * 
	 * @throws Exception
	 *
	 */
	public void start() throws Exception {
		/*
		 * 逻辑判断,先判断启停标识
		 */
		if (revStopSign) {
			throw new Exception("该任务已处于启动状态");
		}
		revStopSign = true;
		producerFutureList = new ArrayList<Future>();
		consumerFutureList = new ArrayList<Future>();
		/*
		 * 创建生产者线程池
		 */
		ThreadFactory producerThreadFactory = new ThreadFactoryBuilder()
				.setNameFormat(config.getKey() + "-produce-pool-%d").build();
		producerExecutor = new ThreadPoolExecutor(this.config.getProducerCount(), this.config.getProducerCount(), 0L,
				TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(1024), producerThreadFactory,
				new ThreadPoolExecutor.AbortPolicy());

		/*
		 * 创建消费者线程池
		 */
		ThreadFactory consumerThreadFactory = new ThreadFactoryBuilder()
				.setNameFormat(config.getKey() + "-consume-pool-%d").build();
		consumerExecutor = new ThreadPoolExecutor(this.config.getConsumerCount(), this.config.getConsumerCount(), 0L,
				TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(1024), consumerThreadFactory,
				new ThreadPoolExecutor.AbortPolicy());

		for (int i = 0; i < config.getProducerCount(); i++) {
			producerFutureList.add(producerExecutor.submit(new Producer()));
		}
		for (int i = 0; i < config.getConsumerCount(); i++) {
			consumerFutureList.add(consumerExecutor.submit(new Consumer()));
		}
	}

	/**
	 * 停止服务
	 * 
	 * @throws Exception
	 */
	public void stop() throws Exception {
		if (!revStopSign) {
			throw new Exception("该任务已处于停止状态");
		}
		revStopSign = false;

		/*
		 * 关闭线程
		 */
		if (!producerFutureList.isEmpty()) {
			for (Future future : producerFutureList) {
				future.cancel(true);
				log.info("生产者关闭线程,Canceld:{}", future.isCancelled());
				log.info("生产者关闭线程,Done:{}", future.isDone());
			}
		}
		if (!consumerFutureList.isEmpty()) {
			for (Future future : consumerFutureList) {
				future.cancel(true);
				log.info("消费者关闭线程,Canceld:{}", future.isCancelled());
				log.info("消费者关闭线程,Done:{}", future.isDone());
			}
		}

		/*
		 * 关闭线程池
		 */
		producerExecutor.shutdown();
		try {
			producerExecutor.awaitTermination(3L, TimeUnit.SECONDS);
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		consumerExecutor.shutdown();
		try {
			consumerExecutor.awaitTermination(3L, TimeUnit.SECONDS);
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	/**
	 * 生产者线程
	 *
	 * @author zhao
	 * @date 2018-11-07
	 */
	class Producer implements Runnable {

		@Override
		public void run() {
			while (revStopSign) {
				List<T> list = Lists.newArrayList();
				try {
					list = produce.produce();
					if (CollectionUtils.isEmpty(list)) {
						// 未查询到数据进行等待 单位millis毫秒
						Thread.sleep(config.getNoDataWaiting() * 1000);
						continue;
					}
				} catch (InterruptedException e) {
					log.debug("生产者等待...");
				} catch (Exception e) {
					log.error(e.getMessage());
				}

				log.info("{}生产数据", Thread.currentThread().getName());
				try {
					for (Iterator<T> iterator = list.iterator(); iterator.hasNext();) {
						queue.put(iterator.next());
					}
				} catch (InterruptedException e) {
					log.debug("生产者生产放入队列失败...");
				} catch (Exception e) {
					log.error(e.getMessage());
				}

			}
		}
	}

	/**
	 * 消费者线程
	 *
	 * @author zhao
	 * @date 2018-11-07
	 */
	class Consumer implements Runnable {

		@Override
		public void run() {
			while (revStopSign) {
				T t = null;
				try {
					t = queue.take();
					log.info("{}消费数据", Thread.currentThread().getName());
					consume.consume(t);
					consume.consumeSuccess(t);
				} catch (InterruptedException e) {
					consume.consumeFail(t);
					log.error(Thread.currentThread().getName() + "-消费数据失败");
				}
			}
		}
	}
}
